package im_server

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"github.com/gin-gonic/gin"
	"github.com/gomodule/redigo/redis"
	"github.com/gorilla/websocket"
	"go.uber.org/atomic"
	"go.uber.org/zap"
	"gorm.io/driver/mysql"
	"gorm.io/gorm"
	"gorm.io/gorm/clause"
	"gorm.io/gorm/logger"
	"log"
	"net/http"
	"os"
	"os/signal"
	pio_im "pim"
	"pim/pkg/codes"
	"pim/pkg/config"
	"pim/pkg/im_server/internal/dao"
	"pim/pkg/models"
	"pim/pkg/promePkg"
	"pim/pkg/tools"
	"sync"
	"syscall"
	"time"
)

type server struct {
	logger          *zap.Logger
	mu              sync.RWMutex
	port            int
	redisPool       *redis.Pool
	closeServer     chan struct{}
	db              *gorm.DB
	config          *config.ImServerConfig
	httpd           *gin.Engine // http 服务
	wsUpGrader      *websocket.Upgrader
	sendMessageChan chan *models.SingleMessageDataType
	saveMessageChan chan *models.SingleMessage

	// 这里的平台 改成 唯一id/流id 即可实现多端多登
	wsConnToUser map[*ClientConn]map[int64]int64 // ws 对象 对应 平台id -> userid
	wsUserToConn map[int64]map[int64]*ClientConn // 用户id 对应 平台id -> ws 对象
	msgNode      *tools.Node
	dbLogger     *DBLogger

	dao dao.APIDao
}

type DBLogger struct {
	logger *zap.Logger
}

func (D DBLogger) Printf(s string, i ...interface{}) {
	//TODO implement me
	//panic("implement me")
	D.logger.Debug("DBLog", zap.String("log", fmt.Sprintf(s, i...)))
}

type SyncClientEvent interface {
	// UpdateUserInfo 有用户信息更新的时候标记这个 , 删除指定client 的用户信息
	UpdateUserInfo(userid int64)
	// UpdateGroupMessage 有群组消息的时候执行这个, 向指定的用户推送群消息
	UpdateGroupMessage(chatID int64, msg *models.SingleMessageDataType)
	//

}

// FileTaskInfo TODO 文件锁
type FileTaskInfo struct {
	Path          string
	Sha256        string
	Name          string
	CurrentOffset int
	FileSize      int
	file          *os.File
	CreateTime    int64
}

func (i *FileTaskInfo) Save() error {
	return i.file.Close()

}

func (i *FileTaskInfo) CheckStatus(currentTime int64) bool {
	// 校验文件是否过期

	deleteTime := i.CreateTime + 200000

	if currentTime > deleteTime {
		i.Delete()
		return true
	}

	return false
}

func (i *FileTaskInfo) Delete() {
	//i.file.
	filePath := fmt.Sprintf("%s/%s", i.Path, i.Name)
	fmt.Println("FileTaskInfo delete ", filePath)
	i.file.Close()
	os.Remove(filePath)

}

type ClientConn struct {
	*websocket.Conn
	IsClose            bool
	ctx                context.Context
	w                  *sync.Mutex
	UserID             int64
	Pf                 int
	IsCompress         bool
	logger             *zap.Logger
	svr                *server
	cacheUserID        map[int64]bool
	cacheChatID        map[int64]*models.WsClientChatInfoCache
	cacheUploadFileMap map[int64]*FileTaskInfo
}

func (c *ClientConn) PushEventToClient(modType int16, subType int16, body interface{}) error {

	var output []byte
	if body != nil {
		output, _ = json.Marshal(body)
	}

	pack := tools.NewProtocolPackage(modType, subType, 0, output)

	outputBytes := pack.Bytes()

	return c.WriteMessage(websocket.BinaryMessage, outputBytes)

}

func (c *ClientConn) UserModAddUserToContact(req *models.Contact) (resp *models.Ok, err error) {

	resp = new(models.Ok)

	// 数据加入数据库

	req.UserID = c.UserID
	//

	// 判断用户是否存在

	var findUser models.UserInfoViewer

	//findUser.UserID = cUserID

	respDB := c.svr.db.Model(&findUser).Where(&models.UserInfoViewer{
		UserID: c.UserID,
	}).Find(&findUser)

	if respDB.Error != nil {
		//err = respDB.Error
		c.logger.Debug("查找用户失败", zap.Error(respDB.Error))
		err = errors.New("添加失败")
		return
	}

	updateMap := map[string]interface{}{}

	//if req.FirstName != "" {
	//	updateMap["first_name"] = req.FirstName
	//}
	//
	//if req.LastName != "" {
	//	updateMap["last_name"] = req.LastName
	//}

	if req.Note != "" {
		updateMap["note"] = req.Note
	}

	timeNow := time.Now()

	updateMap["updated_at"] = timeNow.UnixMilli()

	// 可以添加
	addErr := c.svr.db.Clauses(&clause.OnConflict{
		//DoNothing: false,
		DoUpdates: clause.Assignments(updateMap),
	}).Create(req).Error

	if addErr != nil {
		c.logger.Debug("添加用户失败", zap.Error(addErr))

		err = errors.New("添加失败")
		return
	}

	return
}

func (c *ClientConn) CloseEvent() {
	for _, info := range c.cacheUploadFileMap {
		info.Delete()
	}
}

func (s *server) Run() error {

	// 注册业务

	s.InitHttpApi()

	go s.StartSenderMessageEventService()
	go s.StartSaveMessageEventService()
	// 并入下一个任务
	//go s.dao.AutoClearService()
	go s.StartAutoClearFileTaskService()
	srv := &http.Server{
		Addr:    fmt.Sprintf(":%d", s.port),
		Handler: s.httpd,
	}

	go func() {
		// service connections
		if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
			log.Fatalf("listen: %s\n", err)
		}
		s.logger.Info("http service stop...")

	}()

	quit := make(chan os.Signal)
	// kill (no param) default send syscanll.SIGTERM
	// kill -2 is syscall.SIGINT
	// kill -9 is syscall. SIGKILL but can"t be catch, so don't need add it
	signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
	<-quit
	close(s.closeServer)
	//log.Println("Shutdown Server ...")
	s.logger.Error("Shutdown Server ...")
	ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
	defer cancel()
	if err := srv.Shutdown(ctx); err != nil {
		s.logger.Error("Server Shutdown err :", zap.Error(err))
	}

	select {
	case <-ctx.Done():
		s.logger.Info("timeout of 1 seconds.")
	}
	s.logger.Error("Server exiting")
	return nil
}

func (s *server) RemoveConn(conn *ClientConn, streamID int64) {
	s.mu.Lock()
	defer s.mu.Unlock()
	//operationID := tools.OperationIDGenerator()
	promePkg.PromeInc(promePkg.AccountDisConnCounter)

	var uid int64
	var platform int = conn.Pf
	// 先判断 conn -> user 有没有信息

	// 先找到其他 user id 下的映射

	if streamConnMap, okg := s.wsUserToConn[conn.UserID]; okg {
		// 判断有没有流 也当前相同的 key 删掉

		// 删除流后看有没有数据
		delete(streamConnMap, streamID)

		if len(streamConnMap) == 0 {
			// 删除整个用户关系
			delete(s.wsUserToConn, conn.UserID)
		}
		count := 0
		for _, v := range s.wsUserToConn {
			count = count + len(v)
		}
		s.logger.Debug("user delete ", zap.Int64("uid", uid), zap.Int("pf", platform),
			zap.Int("online_conn_num", count))
	}

	if _, okg := s.wsConnToUser[conn]; okg {
		// stream -> uid
		// 遍历 流id -> user id 的关系

		//

		//for k, v := range oldStringMap {
		//	//platform = k
		//	uid = v
		//}
		//if oldConnMap, ok := s.wsUserToConn[uid]; ok {
		//	delete(oldConnMap, streamID)
		//	s.wsUserToConn[uid] = oldConnMap
		//	if len(oldConnMap) == 0 {
		//		delete(s.wsUserToConn, uid)
		//	}
		//	count := 0
		//	for _, v := range s.wsUserToConn {
		//		count = count + len(v)
		//	}
		//	s.logger.Debug("user delete ", zap.Int64("uid", uid), zap.Int("pf", platform),
		//		zap.Int("online_conn_num", count))
		//} else {
		//	s.logger.Debug("user delete last", zap.Int64("uid", uid), zap.Int("pf", platform))
		//}

		// 清理操作

		for id, info := range conn.cacheUploadFileMap {
			delete(conn.cacheUploadFileMap, id)
			info.Delete()
		}

		delete(s.wsConnToUser, conn)
	}

	if !conn.IsClose {

		err := conn.Close()
		if err != nil {
			s.logger.Error("close err", zap.Int64("streamID", streamID))

		}
	}

}

func (s *server) HttpPostLoginToken(ctx *gin.Context) {
	type ReqDataType struct {
		Account  string `json:"account"`
		Password string `json:"password"`
		Pf       int    `json:"pf"`
	}

	// 查找 用户
	var req ReqDataType
	if err := ctx.BindJSON(&req); err != nil {
		tools.Resp400BadRequest(ctx)
		return
	}

	var userinfo models.Auth

	err := s.db.
		Model(&userinfo).
		Where("(email = ? or username = ? or mobile = ?) and password = ? ",
			req.Account, req.Account, req.Account, req.Password).
		Find(&userinfo).Error
	//err := s.db.Model(&userinfo).Where(&models.Auth{
	//	Email:    req.Email,
	//	Password: req.Password,
	//}).Find(&userinfo).Error

	if userinfo.UserID == 0 {
		tools.Resp500(ctx, "账户或密码错误")

		return
	}
	if err != nil {
		tools.Resp500(ctx, err.Error())
		return
	}

	tokenStr, _ := tools.GenToken(userinfo.UserID, req.Pf, userinfo.Level)

	tools.RespData(ctx, 200, "ok", tokenStr)

	return
}

func (s *server) HttpPostRegisterUser(ctx *gin.Context) {
	type ReqDataType struct {
		Username string `json:"username"`
		Password string `json:"password"`
		Email    string `json:"email"`
	}

	// 直接注册

	var req ReqDataType

	if err := ctx.BindJSON(&req); err != nil {
		tools.Resp400BadRequest(ctx)
		return
	}

	// 创建用户信息

	var authInfo models.Auth

	authInfo.Password = req.Password
	authInfo.Email = req.Email
	authInfo.Username = req.Username

	err := s.db.Create(&authInfo).Error

	if err != nil {
		s.logger.Debug("注册失败", zap.Error(err))
		tools.Resp500(ctx, "注册失败,账户邮箱可能被占用了")
		return
	}

	//// 创建用户信息
	//var userinfo models.UserInfo
	//
	//userinfo.UserID = authInfo.UserID
	//userinfo.Username = req.Username
	//
	//err = s.db.Create(&userinfo).Error
	//
	//if err != nil {
	//	tools.Resp500(ctx, err.Error())
	//	return
	//}
	tools.RespData(ctx, 200, "ok", authInfo.UserID)
	return

	//		m := gomail.NewMessage()
	//		m.SetHeader(`From`, config.Config.Demo.Mail.SenderMail)
	//		m.SetHeader(`To`, []string{account}...)
	//		m.SetHeader(`Subject`, config.Config.Demo.Mail.Title)
	//		m.SetBody(`text/html`, fmt.Sprintf("%d", code))
	//		if err := gomail.NewDialer(config.Config.Demo.Mail.SmtpAddr, config.Config.Demo.Mail.SmtpPort, config.Config.Demo.Mail.SenderMail, config.Config.Demo.Mail.SenderAuthorizationCode).DialAndSend(m); err != nil {
	//			log.Error(params.OperationID, "send mail error", account, err.Error())
	//			c.JSON(http.StatusOK, gin.H{"errCode": constant.MailSendCodeErr, "errMsg": ""})
	//			return
	//		}
}

func (s *server) StartSenderMessageEventService() {

	// 处理发送消息的数据

	procSenderMsg := func(msg *models.SingleMessageDataType) {
		defer tools.HandlePanic(s.logger, "StartSenderMessageEventServiceProcMessage")

		// 转发到指定的用户
		// 新消息事件

		if msg.ChatID < 0 {
			// 这是群

			// TODO 群处理
			targetChatType := msg.ChatID & 0xff0000000000 // 群标志
			if targetChatType == 0x010000000000 {
				// 普通群 -- 200 人以下规模 私有

			} else if targetChatType == 0x100000000000 {
				// 超级群 -- 100000 人规模 公开

			}

		} else {
			// 这是个人
			// 查找 个人id
			// 查找用户id

			localUserID := msg.ChatID & 0xffff

			// 搜索网络的用户

			recvUserConn := s.wsUserToConn[localUserID]
			myUserConn := s.wsUserToConn[msg.Sender]

			// 接收的用户id
			// 将我的消息

			msg.MsgStatus = models.SenderMsgStateSuccess

			mySenderBuffer, _ := json.Marshal(msg)

			eventPackage := tools.ProtocolPackage{
				Body:    mySenderBuffer,
				Version: 1,
				ModType: codes.EventMod,
				SubType: codes.EventModNewMessage,
			}
			// 新消息事件
			senderData := eventPackage.Bytes()

			for streamID, conn := range myUserConn {
				s.logger.Info("推送消息事件给多端", zap.Int64("streamID", streamID), zap.Int64("chat_id", msg.ChatID), zap.Int64("Sender", msg.Sender), zap.Int64("msg_id", msg.MsgID))
				conn.WriteMessage(websocket.BinaryMessage, senderData)
			}

			// TODO 离线消息处理
			// 接收方需要改写 chat id
			if msg.ChatID == localUserID {
				msg.ChatID = msg.Sender
			}

			otherSenderBuffer, _ := json.Marshal(msg)

			otherEventPackage := tools.ProtocolPackage{
				Body:    otherSenderBuffer,
				Version: 1,
				ModType: codes.EventMod,
				SubType: codes.EventModNewMessage,
			}
			// 新消息事件
			otherSenderData := otherEventPackage.Bytes()

			for streamID, conn := range recvUserConn {
				s.logger.Info("推送消息事件给用户", zap.Int64("streamID", streamID), zap.Int64("chat_id", msg.ChatID), zap.Int64("Sender", msg.Sender), zap.Int64("msg_id", msg.MsgID))
				// 判断接收的用户有没有缓存这个发送者

				canSender := conn.SenderMsgCheck(msg.Sender, msg.Sender, false)
				// 判断是否可以发送 , 不可以发送则跳过
				if !canSender {
					continue
				}
				conn.WriteMessage(websocket.BinaryMessage, otherSenderData)
			}

		}

	}

	for {

		select {
		case item := <-s.sendMessageChan:

			// 处理发送的消息
			procSenderMsg(item)
		case <-s.closeServer:
			s.logger.Warn("StartSenderMessageEventService stop")
			return
		}
	}

}

func (s *server) StartSaveMessageEventService() {
	ticker := time.NewTicker(time.Second * 10)

	var saveMessageItems []*models.SingleMessage
	var saveStart = atomic.NewBool(false)

	doSaveMessage := func(timeNow time.Time) {
		if len(saveMessageItems) == 0 {
			return
		}

		if saveStart.Load() {
			// 有任务再保存
			return
		}

		saveStart.Store(true)
		defer saveStart.Store(false)

		// 保存数据

		err := s.db.Create(saveMessageItems).Error
		if err != nil {
			s.logger.Debug("Save info ", zap.Int("size", len(saveMessageItems)))
		}

		saveMessageItems = saveMessageItems[0:0]

	}

	for true {
		select {
		case timeNow := <-ticker.C:
			doSaveMessage(timeNow)

		case item := <-s.saveMessageChan:
			saveMessageItems = append(saveMessageItems, item)

		case <-s.closeServer:
			s.logger.Warn("StartSaveMessageEventService stop")
			doSaveMessage(time.Now())
			return

		}
	}

}

func (s *server) GetWriter() logger.Writer {
	return s.dbLogger
}

func New(config *config.ImServerConfig) pio_im.Runner {

	svr := &server{
		config:          config,
		wsConnToUser:    make(map[*ClientConn]map[int64]int64),
		wsUserToConn:    make(map[int64]map[int64]*ClientConn),
		sendMessageChan: make(chan *models.SingleMessageDataType, 16),
		saveMessageChan: make(chan *models.SingleMessage, 16),
		closeServer:     make(chan struct{}),
	}

	svr.port = config.HttpPort

	var err error
	svr.logger = zap.NewExample()

	svr.dbLogger = &DBLogger{
		logger: svr.logger,
	}
	svr.redisPool, err = tools.NewRedis(config.RedisIP, config.RedisPassword, config.RedisDB)

	if err != nil {
		panic(err)
	}

	db, err := gorm.Open(mysql.New(mysql.Config{
		DSN: config.DBUri,
		//DefaultStringSize: 256, // string 类型字段的默认长度
		//DisableDatetimePrecision: true, // 禁用 datetime 精度，MySQL 5.6 之前的数据库不支持
		//DontSupportRenameIndex: true, // 重命名索引时采用删除并新建的方式，MySQL 5.7 之前的数据库和 MariaDB 不支持重命名索引
		//DontSupportRenameColumn: true, // 用 `change` 重命名列，MySQL 8 之前的数据库和 MariaDB 不支持重命名列
		SkipInitializeWithVersion: true, // 根据当前 MySQL 版本自动配置
	}), &gorm.Config{
		Logger: logger.New(svr.GetWriter(), logger.Config{
			LogLevel: logger.Info,
		}),
	})
	if err != nil {
		panic(err)
	}

	svr.db = db

	svr.dao = dao.NewDao(svr.logger, db, svr.redisPool, svr.closeServer)
	svr.msgNode, err = tools.NewNode(int64(1))

	if err != nil {
		panic(err)
	}

	svr.httpd = gin.Default()

	var upgrader = &websocket.Upgrader{
		// 解决跨域问题
		CheckOrigin: func(r *http.Request) bool {
			return true
		},
	} // use default options
	svr.wsUpGrader = upgrader
	InitDBTables(db)

	promePkg.InitConter()

	return svr
}

func InitDBTables(db *gorm.DB) {
	// TODO 初始化建表

	var mig gorm.Migrator
	dbType := db.Config.Name()

	if dbType == "mysql" {
		mig = db.Set("gorm:table_options", "ENGINE = InnoDB DEFAULT CHARSET=utf8mb4").Migrator()
	} else {
		mig = db.Migrator()
	}

	_ = mig.AutoMigrate(&models.SingleMessage{})
	//_ = db.AutoMigrate(&models.UserInfo{})
	_ = mig.AutoMigrate(&models.Auth{})
	_ = mig.AutoMigrate(&models.ChatInfo{})
	_ = mig.AutoMigrate(&models.Contact{})
	_ = mig.AutoMigrate(&models.FileUUID{})
}
